package org.raymond.iworks.study.mq.rabbitmq.classic.rpc;

import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;

/**
 * @author raymond
 * @version V1.0
 * @Description: TODO
 */
@Slf4j
public class RPCClient implements AutoCloseable {

    /**
     * 1. 对于rpc请求,客户端需要发送两个属性:
     * reply_to, 仅为请求创建的匿名独占队列
     * correlation_id, 被设置为每个请求的唯一值
     * 2. 请求被发送到rpc_queue队列
     * 3. rpc服务器在该队列上等待请求,当一个请求到达时,服务器完成任务,
     * 将响应报文写入reply_to上的回复队列
     * 4. rpc客户端等待reply_to队列上的响应报文.当响应到达时,rpc客户端
     * 会检查correlationId属性.如果它与请求中的值匹配,则将响应返回给应用程序
     * @param args
     */
    private Connection connection;
    private Channel channel;

    public RPCClient() throws Exception{
        ConnectionFactory factory = factory();
        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public String call(String message) throws Exception {
        final String corrId = UUID.randomUUID().toString();

        // 声明一个匿名独占自动删除的瞬态队列,用于存储服务器响应的报文
        String replyQueueName = channel.queueDeclare().getQueue();
        AMQP.BasicProperties props = new AMQP.BasicProperties()
                .builder()
                //.expiration("6000")  // 消息的过期时间,队列也有设置时,以较小那个为准
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();
        // rpc模式,不需要交换机
        // RPCServer.RPC_QUEUE_NAME, 请求队列,用于存储向服务器发送的请求报文
        channel.basicPublish("", RPCServer.RPC_QUEUE_NAME, props, message.getBytes(StandardCharsets.UTF_8));

        final CompletableFuture<String> response = new CompletableFuture<>();

        boolean autoAck = false;
        String ctag = channel.basicConsume(replyQueueName, autoAck, (consumerTag, delivery)->{
            try {
                if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                    response.complete(new String(delivery.getBody(), StandardCharsets.UTF_8));
                }
            }finally {
                channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
            }
        }, consumerTag->{});

        String result = response.get();
        channel.basicCancel(ctag); // consumerTag,取消消费者对队列的订阅关系
        return result;
    }

    public void close() throws IOException {
        connection.close();
    }
    public static void main(String[] args) {
        try(RPCClient fibRpc = new RPCClient()){
            for(int i=0; i<32; i++){
                String iStr = Integer.toString(i);
                log.info(" [x] Requesting fib ({})", i);
                String response = fibRpc.call(iStr);
                log.info(" [.] Got {}", response);
            }
        } catch (Exception e) {
            log.error("", e);
        }
    }

    private static ConnectionFactory factory(){
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.125");
        factory.setPort(5672);
        return factory;
    }
}
